const { parentPort } = require('worker_threads')
const tables = {}
global.tables = tables
global.Acan = require('../lib/Acan')
global._inc = Acan.random(16, 99)
require('../lib/Adebug')
const fs = require('fs')
const Fn = require('./fn.js')
const Adb = require('./mod.js')
const psend = parentPort ? parentPort.postMessage : (arr) => { process.send(arr) }

process.on('unhandledRejection', (reason, p) => {
  console.log(process.pid, reason, p)
})
// 正式环境的开启错误捕捉，防止进程挂掉
process.on('uncaughtException', function(err) {
  console.log(process.pid, err)
})

// opt.hotTime 热数据时间（分钟）hotLimit 热数据数量限制 只有热数据才能做修改
// opt.update 是否允许数据更新
const createTable = async(ns, cols, opt = { update: false }, isInit = 0) => {
  opt.tableMem = opt.tableMem || conf.tableMem
  const nrr = ns.split('.')
  if (nrr.length !== 2) return { code: 101, message: 'error ns' }
  const defCols = opt.update ? { _id: { len: 12, type: '_id', encoding: 'hex' }} : { _t: { len: 4, type: 'time', encoding: 'hex' }}
  if (!isInit) cols = Object.assign(defCols, cols) // 时间索引 or 对象ID索引 系统定义
  const o = { ns, db: nrr[0], name: nrr[1], inc: 0, incs: {}, opt, cols, fd: {}, hourList: [], _sab: {}, _buf: {}} // inc 计数器
  try {
    o.hourList = JSON.parse(fs.readFileSync(`${conf.db}/${o.db}/${o.name}/list.json`, { encoding: 'utf-8' })) || []
  } catch (error) {}
  let unitLen = 0
  for (const k in cols) {
    const col = cols[k]
    if (!isInit) {
      if (!col.len) col.len = { bool: 1, float: 4, double: 8, time: 4 }[col.type]
      col.len += 1 // 多一个字节用于 $eq $ne
    }
    unitLen += col.len
  }
  opt.hotLimit = parseInt(opt.tableMem / unitLen * 1024 * 1.024) * 1000
  o.memSize = 0
  for (const k in cols) {
    await Fn.snapShotLoad(o, k, cols[k], opt, isInit) // 加载老的列数据
    o.memSize += o._buf[k].byteLength
  }
  Fn.log(`creatTable ${ns} ok`)
  Object.assign(o, Adb)
  tables[ns] = o
  o.saveTableInfo()
  return o
}

global.caches = {
  add(hash, val) {
    this.data[hash] = val
    if (!this.hash[this.hour]) this.hash[this.hour] = []
    this.hash[this.hour].push(hash)
  },
  data: {},
  hash: {},
  hour: Acan.time('YmdH')
}

const actCount = { create: { inc: 0 } }
let last
const rate = {}
let lastCpu = process.cpuUsage()
const cpuRate = {}
setInterval(() => {
  // 计算每秒CPU使用率
  const nowCpu = process.cpuUsage()
  cpuRate.user = nowCpu.user - lastCpu.user
  cpuRate.system = nowCpu.system - lastCpu.system
  lastCpu = nowCpu
  // 计算每秒请求量
  const now = {}
  Acan.each(actCount, (fn, k) => { now[k] = fn.inc })
  if (last) {
    Acan.each(now, (v, k) => {
      if (!v || !last[k]) return
      rate[k] = v - last[k]
    })
    // console.log(process.pid, rate)
  }
  last = now
  // 同步子进程统计数据
  const rs = Object.assign({ total: { table: 0, inc: 0, memSize: 0 }, mem: process.memoryUsage() })
  Acan.each(tables, (table) => {
    rs.total.table++
    rs.total.inc += table.inc
    rs.total.memSize += table.memSize
  })
  psend([2, 'qps', process.pid, rs])
}, 1000)

setInterval(() => {
  caches.hour = Acan.time('YmdH')
}, 60000) // 每分钟任务
setInterval(() => {
  const preHour = Acan.time('YmdH', Acan.time() - 3600)
  if (caches.hash[preHour]) {
    caches.hash[preHour].map(hash => {
      delete caches.data[hash]
    })
    delete caches.hash[preHour]
  }
}, 3600000) // 每小时 删除上一小时的缓存

const startTime = Acan.time() // 进程启动时间
const mod = {
  async init() {
    let i = 0
    Acan.mkdirs(conf.db + '/log')
    for (const op of fs.readdirSync(conf.db, { withFileTypes: true })) {
      if (!op.isDirectory()) continue
      const dbPath = conf.db + '/' + op.name
      for (const o of fs.readdirSync(dbPath, { withFileTypes: true })) {
        if (!o.isDirectory()) continue
        const namePath = dbPath + '/' + o.name
        let tableConf = {}
        try {
          Object.assign(tableConf, JSON.parse(fs.readFileSync(namePath + '/conf.json')))
        } catch (error) {}
        if (tableConf.ns) {
          i++
          Adebug.log(`init table: ${tableConf.ns}`)
          await createTable(tableConf.ns, tableConf.cols, tableConf.opt, 1)
        }
      }
    }
    Fn.log(`init tables: ${i}`)
  },
  async conf([id, m, o]) {
    global.conf = o
    await mod.init()
    psend([0, 'init', 'ok'])
  },
  cpu([id]) {
    psend([id, 'cpu', Object.assign({ pid: process.pid }, cpuRate)])
  },
  async restart([id]) {
    const rs = await Fn.snapShot()
    psend([id, 'restart', rs])
  },
  async createTable([id, m, ns, cols, opt]) {
    await createTable(ns, cols, opt)
    psend([id, 'createTable', {}])
  },
  initTable(arr) {
    const table = arr[2]
    table._buf = {}
    let i = 4
    for (const k of arr[3]) {
      table._buf[k] = Buffer.from(arr[i])
      i++
    }
    Object.assign(table, Adb)
    tables[table.ns] = table
  },
  tableList([id, m, get]) {
    const rs = { startTime, pid: process.pid, uptime: process.uptime(), total: { table: 0, inc: 0, memSize: 0 }, tables: []}
    Acan.each(tables, (table, ns) => {
      if (get.db && table.db !== get.db) return
      const o = { ns, db: table.db, name: table.name, inc: table.inc, memSize: table.memSize, overMoveLock: table.overMoveLock, opt: table.opt, cols: Acan.clone(table.cols) }
      Acan.each(table._buf, (buf, k) => {
        o.cols[k].memSize = buf.length
      })
      rs.total.table++
      rs.total.inc += o.inc
      rs.total.memSize += o.memSize
      rs.tables.push(o)
    })
    psend([id, 'tableList', rs])
  },
  async table([id, m, act, ns, get, opt]) {
    if (!actCount[act]) actCount[act] = { inc: 0 }
    if (act === 'creates') actCount.create.inc += get.length
    const table = tables[ns]
    if (!table) return psend([id, act, { code: 400, pid: process.pid, message: `not found ${ns}` }])
    if (!table[act]) return psend([id, act, { code: 400, pid: process.pid, message: `not found ${act}` }])
    // if (act === 'creates') return psend([id, act, { count: get.length }])
    psend([id, act, await table[act](get, opt)])
    actCount[act].inc++
  },
}

process.on('message', arr => {
  const fn = mod[arr[1]]
  if (!fn) return Adebug.log('thread no fn!', arr)
  fn(arr)
})

const main = async () => {
  Adebug.info('adb fork', process.pid)
}
main()
